{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "14d78fa1-a9b9-41f7-9285-7d27048ecb05",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import findspark\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "51ad4495-24bf-4954-b527-08efc22887af",
   "metadata": {},
   "outputs": [],
   "source": [
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "83568678-b69e-4ff2-9c03-f9f26ec77188",
   "metadata": {},
   "outputs": [],
   "source": [
    "folder_path = r'DataAnalysisWithPythonAndPySpark-Data-trunk\\broadcast_logs'\n",
    "csv_path = os.path.join(folder_path, r'BroadcastLogs_2018_Q3_M8_sample.CSV')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "c6dab87e-5bf6-4175-83e6-ef1dd3bd58f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = (\n",
    "    SparkSession\n",
    "    .builder\n",
    "    .appName('brocast analysis')\n",
    "    .getOrCreate()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e3a1261c-8164-4b9c-be01-b829b6ce2fe8",
   "metadata": {},
   "outputs": [],
   "source": [
    "logs = spark.read.csv(csv_path,\n",
    "                     sep = '|',\n",
    "                     header = True,\n",
    "                     inferSchema = True,\n",
    "                     timestampFormat = 'yyyy-MM-dd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "9bbffcd2-fdad-403f-968c-7b334b3531de",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on method csv in module pyspark.sql.readwriter:\n",
      "\n",
      "csv(path: Union[str, List[str]], schema: Union[pyspark.sql.types.StructType, str, NoneType] = None, sep: Optional[str] = None, encoding: Optional[str] = None, quote: Optional[str] = None, escape: Optional[str] = None, comment: Optional[str] = None, header: Union[bool, str, NoneType] = None, inferSchema: Union[bool, str, NoneType] = None, ignoreLeadingWhiteSpace: Union[bool, str, NoneType] = None, ignoreTrailingWhiteSpace: Union[bool, str, NoneType] = None, nullValue: Optional[str] = None, nanValue: Optional[str] = None, positiveInf: Optional[str] = None, negativeInf: Optional[str] = None, dateFormat: Optional[str] = None, timestampFormat: Optional[str] = None, maxColumns: Union[str, int, NoneType] = None, maxCharsPerColumn: Union[str, int, NoneType] = None, maxMalformedLogPerPartition: Union[str, int, NoneType] = None, mode: Optional[str] = None, columnNameOfCorruptRecord: Optional[str] = None, multiLine: Union[bool, str, NoneType] = None, charToEscapeQuoteEscaping: Optional[str] = None, samplingRatio: Union[str, float, NoneType] = None, enforceSchema: Union[bool, str, NoneType] = None, emptyValue: Optional[str] = None, locale: Optional[str] = None, lineSep: Optional[str] = None, pathGlobFilter: Union[bool, str, NoneType] = None, recursiveFileLookup: Union[bool, str, NoneType] = None, modifiedBefore: Union[bool, str, NoneType] = None, modifiedAfter: Union[bool, str, NoneType] = None, unescapedQuoteHandling: Optional[str] = None) -> 'DataFrame' method of pyspark.sql.readwriter.DataFrameReader instance\n",
      "    Loads a CSV file and returns the result as a  :class:`DataFrame`.\n",
      "    \n",
      "    This function will go through the input once to determine the input schema if\n",
      "    ``inferSchema`` is enabled. To avoid going through the entire data once, disable\n",
      "    ``inferSchema`` option or specify the schema explicitly using ``schema``.\n",
      "    \n",
      "    .. versionadded:: 2.0.0\n",
      "    \n",
      "    .. versionchanged:: 3.4.0\n",
      "        Supports Spark Connect.\n",
      "    \n",
      "    Parameters\n",
      "    ----------\n",
      "    path : str or list\n",
      "        string, or list of strings, for input path(s),\n",
      "        or RDD of Strings storing CSV rows.\n",
      "    schema : :class:`pyspark.sql.types.StructType` or str, optional\n",
      "        an optional :class:`pyspark.sql.types.StructType` for the input schema\n",
      "        or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).\n",
      "    \n",
      "    Other Parameters\n",
      "    ----------------\n",
      "    Extra options\n",
      "        For the extra options, refer to\n",
      "        `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-csv.html#data-source-option>`_\n",
      "        for the version you use.\n",
      "    \n",
      "        .. # noqa\n",
      "    \n",
      "    Examples\n",
      "    --------\n",
      "    Write a DataFrame into a CSV file and read it back.\n",
      "    \n",
      "    >>> import tempfile\n",
      "    >>> with tempfile.TemporaryDirectory() as d:\n",
      "    ...     # Write a DataFrame into a CSV file\n",
      "    ...     df = spark.createDataFrame([{\"age\": 100, \"name\": \"Hyukjin Kwon\"}])\n",
      "    ...     df.write.mode(\"overwrite\").format(\"csv\").save(d)\n",
      "    ...\n",
      "    ...     # Read the CSV file as a DataFrame with 'nullValue' option set to 'Hyukjin Kwon'.\n",
      "    ...     spark.read.csv(d, schema=df.schema, nullValue=\"Hyukjin Kwon\").show()\n",
      "    +---+----+\n",
      "    |age|name|\n",
      "    +---+----+\n",
      "    |100|NULL|\n",
      "    +---+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(spark.read.csv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "22a98968-f861-4ab6-9b82-f3932875ec48",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- BroadcastLogID: integer (nullable = true)\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- SequenceNO: integer (nullable = true)\n",
      " |-- AudienceTargetAgeID: integer (nullable = true)\n",
      " |-- AudienceTargetEthnicID: integer (nullable = true)\n",
      " |-- CategoryID: integer (nullable = true)\n",
      " |-- ClosedCaptionID: integer (nullable = true)\n",
      " |-- CountryOfOriginID: integer (nullable = true)\n",
      " |-- DubDramaCreditID: integer (nullable = true)\n",
      " |-- EthnicProgramID: integer (nullable = true)\n",
      " |-- ProductionSourceID: integer (nullable = true)\n",
      " |-- ProgramClassID: integer (nullable = true)\n",
      " |-- FilmClassificationID: integer (nullable = true)\n",
      " |-- ExhibitionID: integer (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- EndTime: string (nullable = true)\n",
      " |-- LogEntryDate: date (nullable = true)\n",
      " |-- ProductionNO: string (nullable = true)\n",
      " |-- ProgramTitle: string (nullable = true)\n",
      " |-- StartTime: string (nullable = true)\n",
      " |-- Subtitle: string (nullable = true)\n",
      " |-- NetworkAffiliationID: integer (nullable = true)\n",
      " |-- SpecialAttentionID: integer (nullable = true)\n",
      " |-- BroadcastOriginPointID: integer (nullable = true)\n",
      " |-- CompositionID: integer (nullable = true)\n",
      " |-- Producer1: string (nullable = true)\n",
      " |-- Producer2: string (nullable = true)\n",
      " |-- Language1: integer (nullable = true)\n",
      " |-- Language2: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "logs.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "80ecb8a4-6ce4-4d28-8ae9-4fd9e0e98521",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------+----------+\n",
      "|BroadcastLogID|LogServiceID|LogDate   |\n",
      "+--------------+------------+----------+\n",
      "|1196192316    |3157        |2018-08-01|\n",
      "|1196192317    |3157        |2018-08-01|\n",
      "|1196192318    |3157        |2018-08-01|\n",
      "|1196192319    |3157        |2018-08-01|\n",
      "|1196192320    |3157        |2018-08-01|\n",
      "+--------------+------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "logs.select(['BroadcastLogID','LogServiceID','LogDate']).show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "7b54be9d-2714-454e-b9d4-ebd594a8bb39",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "list"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(logs.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "1ede5a1c-8895-4e73-ba1a-28c77495eb1e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('BroadcastLogID', 'int'),\n",
       " ('LogServiceID', 'int'),\n",
       " ('LogDate', 'date'),\n",
       " ('SequenceNO', 'int'),\n",
       " ('AudienceTargetAgeID', 'int'),\n",
       " ('AudienceTargetEthnicID', 'int'),\n",
       " ('CategoryID', 'int'),\n",
       " ('ClosedCaptionID', 'int'),\n",
       " ('CountryOfOriginID', 'int'),\n",
       " ('DubDramaCreditID', 'int'),\n",
       " ('EthnicProgramID', 'int'),\n",
       " ('ProductionSourceID', 'int'),\n",
       " ('ProgramClassID', 'int'),\n",
       " ('FilmClassificationID', 'int'),\n",
       " ('ExhibitionID', 'int'),\n",
       " ('Duration', 'string'),\n",
       " ('EndTime', 'string'),\n",
       " ('LogEntryDate', 'date'),\n",
       " ('ProductionNO', 'string'),\n",
       " ('ProgramTitle', 'string'),\n",
       " ('StartTime', 'string'),\n",
       " ('Subtitle', 'string'),\n",
       " ('NetworkAffiliationID', 'int'),\n",
       " ('SpecialAttentionID', 'int'),\n",
       " ('BroadcastOriginPointID', 'int'),\n",
       " ('CompositionID', 'int'),\n",
       " ('Producer1', 'string'),\n",
       " ('Producer2', 'string'),\n",
       " ('Language1', 'int'),\n",
       " ('Language2', 'int')]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logs.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "458b1b50-adf0-49a8-a54a-05c6f55ee3c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "logs = logs.withColumn(\n",
    "    \"duration_seconds\",\n",
    "    (\n",
    "        F.col(\"Duration\").substr(1,2).cast(\"int\")*60*60\n",
    "        +  F.col(\"Duration\").substr(4,2).cast(\"int\")*60\n",
    "        +  F.col(\"Duration\").substr(7,2).cast(\"int\")\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "7de6f9c9-cf45-43f9-b9d5-552f8e15b1b6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                 _c0|\n",
      "+--------------------+\n",
      "|BroadcastLogID|Lo...|\n",
      "|1196192316|3157|2...|\n",
      "|1196192317|3157|2...|\n",
      "|1196192318|3157|2...|\n",
      "|1196192319|3157|2...|\n",
      "+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "logs_raw = spark.read.csv(csv_path)\n",
    "logs_raw.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "7ea2a8f9-c7c7-44e0-82c9-40abb9ae1b7d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "logs_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "134f86d7-ea3e-4c2d-9d8d-568db513b640",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- SequenceNO: integer (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- EndTime: string (nullable = true)\n",
      " |-- LogEntryDate: date (nullable = true)\n",
      " |-- ProductionNO: string (nullable = true)\n",
      " |-- ProgramTitle: string (nullable = true)\n",
      " |-- StartTime: string (nullable = true)\n",
      " |-- Subtitle: string (nullable = true)\n",
      " |-- Producer1: string (nullable = true)\n",
      " |-- Producer2: string (nullable = true)\n",
      " |-- Language1: integer (nullable = true)\n",
      " |-- Language2: integer (nullable = true)\n",
      " |-- duration_seconds: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    logs.select([col for col in logs.columns if not col.endswith(\"ID\")])\n",
    "    .printSchema()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4676f44f-ee44-4994-bd76-9dcc0fca7a70",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- LogIdentifierID: string (nullable = true)\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- PrimaryFG: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csv_path = os.path.join(folder_path,r'ReferenceTables\\LogIdentifier.csv')\n",
    "log_identifier = spark.read.csv(csv_path,\n",
    "                                sep = '|',\n",
    "                                 header = True,\n",
    "                                 inferSchema = True,\n",
    "                                 timestampFormat = 'yyyy-MM-dd')\n",
    "log_identifier.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "2013f11f-adb2-4d1b-b565-25407191395b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|PrimaryFG|\n",
      "+---------+\n",
      "|        1|\n",
      "|        0|\n",
      "+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_identifier.select('PrimaryFG').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "aee628fd-4a29-4bcd-b090-0feca9a99e95",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "920"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "log_identifier.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "b38db065-70cf-4d97-84a3-4e51a8cbb5fc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "758"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "log_identifier = log_identifier.where(F.col('PrimaryFG')==1)\n",
    "log_identifier.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "7f23fe96-fa11-47e1-a964-987356f771d6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+------------+---------+\n",
      "|LogIdentifierID|LogServiceID|PrimaryFG|\n",
      "+---------------+------------+---------+\n",
      "|           13ST|        3157|        1|\n",
      "|         2000SM|        3466|        1|\n",
      "|           70SM|        3883|        1|\n",
      "|           80SM|        3590|        1|\n",
      "|           90SM|        3470|        1|\n",
      "+---------------+------------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_identifier.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "9863aaef-81ce-48d9-a534-56d29e4638ab",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(log_identifier)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "313c7097-d0e8-4bd1-8670-8f2dc75a0e89",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "758"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "log_identifier.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "a55f17bd-84f4-4118-a1a6-26955c5a9cb0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "238945"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logs.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "9e9d92a9-b3b9-4e7d-a22f-7da8079f0f38",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------------+----------+---------------+---------+\n",
      "|LogServiceID|BroadcastLogID|   LogDate|LogIdentifierID|PrimaryFG|\n",
      "+------------+--------------+----------+---------------+---------+\n",
      "|        3157|    1196192320|2018-08-01|           13ST|        1|\n",
      "|        3157|    1196192319|2018-08-01|           13ST|        1|\n",
      "|        3157|    1196192318|2018-08-01|           13ST|        1|\n",
      "|        3157|    1196192317|2018-08-01|           13ST|        1|\n",
      "|        3157|    1196192316|2018-08-01|           13ST|        1|\n",
      "+------------+--------------+----------+---------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "left_table = logs.select(['BroadcastLogID','LogServiceID','LogDate']).limit(5)\n",
    "right_table = log_identifier.limit(5)\n",
    "join_table = left_table.join(right_table,\n",
    "               on = \"LogServiceID\",\n",
    "               how = \"inner\")\n",
    "join_table.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "cf1a3903-e663-4f9e-8f17-5559cb40740e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- BroadcastLogID: integer (nullable = true)\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- LogIdentifierID: string (nullable = true)\n",
      " |-- PrimaryFG: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "join_table.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "04728b8c-c948-4568-afef-06ba36da211e",
   "metadata": {},
   "source": [
    "取前5行进行联结"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "2ca803d1-5937-44cc-b037-5789f0ed660b",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------+----------+---------------+------------+---------+\n",
      "|BroadcastLogID|LogServiceID|   LogDate|LogIdentifierID|LogServiceID|PrimaryFG|\n",
      "+--------------+------------+----------+---------------+------------+---------+\n",
      "|    1196192320|        3157|2018-08-01|           13ST|        3157|        1|\n",
      "|    1196192319|        3157|2018-08-01|           13ST|        3157|        1|\n",
      "|    1196192318|        3157|2018-08-01|           13ST|        3157|        1|\n",
      "|    1196192317|        3157|2018-08-01|           13ST|        3157|        1|\n",
      "|    1196192316|        3157|2018-08-01|           13ST|        3157|        1|\n",
      "+--------------+------------+----------+---------------+------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "left_table.join(right_table,\n",
    "               on = left_table[\"LogServiceID\"]==right_table[\"LogServiceID\"],\n",
    "               how = \"inner\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "9cf1028a-0e29-495f-adc3-4eaeddc0110c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------+----------+\n",
      "|BroadcastLogID|LogServiceID|   LogDate|\n",
      "+--------------+------------+----------+\n",
      "|    1196192316|        3157|2018-08-01|\n",
      "|    1196192317|        3157|2018-08-01|\n",
      "|    1196192318|        3157|2018-08-01|\n",
      "|    1196192319|        3157|2018-08-01|\n",
      "|    1196192320|        3157|2018-08-01|\n",
      "+--------------+------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "left_table.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "8dd071c4-6524-463b-864d-666bd1f644c0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+------------+---------+\n",
      "|LogIdentifierID|LogServiceID|PrimaryFG|\n",
      "+---------------+------------+---------+\n",
      "|         2000SM|        3466|        1|\n",
      "|           70SM|        3883|        1|\n",
      "|           80SM|        3590|        1|\n",
      "|           90SM|        3470|        1|\n",
      "+---------------+------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "right_table2 = right_table.where(\"LogServiceID<>3157\")\n",
    "right_table2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "e572823f-57f0-44d4-b116-8c32acbebc09",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------------+----------+\n",
      "|LogServiceID|BroadcastLogID|   LogDate|\n",
      "+------------+--------------+----------+\n",
      "|        3157|    1196192316|2018-08-01|\n",
      "|        3157|    1196192317|2018-08-01|\n",
      "|        3157|    1196192318|2018-08-01|\n",
      "|        3157|    1196192319|2018-08-01|\n",
      "|        3157|    1196192320|2018-08-01|\n",
      "+------------+--------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "left_table.join(right_table2,\n",
    "               on = \"LogServiceID\",\n",
    "               how = \"left_anti\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "db23583e-7aa2-4269-861e-3437d45a3d3f",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_and_channels = logs.join(log_identifier,\n",
    "                            on = \"LogServiceID\",\n",
    "                            how = \"inner\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "43ab9fad-e8ca-414f-a2ec-d832e2094ec7",
   "metadata": {},
   "outputs": [],
   "source": [
    "csv_path = os.path.join(folder_path,\"ReferenceTables\\CD_Category.csv\")\n",
    "cd_category = spark.read.csv(csv_path,\n",
    "                            sep = '|',\n",
    "                            header = True,\n",
    "                            inferSchema = True\n",
    "                            ).select(\"CategoryID\",\"CategoryCD\",F.col(\"EnglishDescription\").alias(\"Category_Description\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "36166bb4-752f-496d-b003-2125c13d66d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "csv_path = os.path.join(folder_path,\"ReferenceTables\\CD_ProgramClass.csv\")\n",
    "cd_program = spark.read.csv(csv_path,\n",
    "                            sep = '|',\n",
    "                            header = True,\n",
    "                            inferSchema = True\n",
    "                            ).select(\"ProgramClassID\",\"ProgramClassCD\",F.col(\"EnglishDescription\").alias(\"ProgramClass_Description\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "d3de04a2-3aeb-4130-aa33-c6d4896d129f",
   "metadata": {},
   "outputs": [],
   "source": [
    "full_log = log_and_channels.join(cd_category,\"CategoryID\",\"left\").join(cd_program,\"ProgramClassID\",\"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "b09c324f-0e99-4477-a6e6-25afac05a6e4",
   "metadata": {},
   "outputs": [],
   "source": [
    "temp = (\n",
    "    full_log.groupby(\"ProgramClassCD\",\"ProgramClass_Description\")\n",
    "    .agg(F.sum(\"duration_seconds\").alias(\"duration_total\"))\n",
    "    .orderBy(\"duration_total\",ascending=False)\n",
    "    # .show(10,False)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "5c274fdd-d9e9-41a6-a531-f38827c0d98d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(temp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "90bd3f4e-2f09-428e-8d62-9cdae9e13e92",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-------------------+--------------+------------------+\n",
      "|LogIdentifierID|duration_commercial|duration_total|  commercial_ratio|\n",
      "+---------------+-------------------+--------------+------------------+\n",
      "|           CIMT|                775|           775|               1.0|\n",
      "|           MSET|               2700|          2700|               1.0|\n",
      "|          TLNSP|              15480|         15480|               1.0|\n",
      "|         TELENO|              17790|         17790|               1.0|\n",
      "|          HPITV|                 13|            13|               1.0|\n",
      "|           TANG|               8125|          8125|               1.0|\n",
      "|           MMAX|              23333|         23582|0.9894410991434145|\n",
      "|           MPLU|              20587|         20912|0.9844586840091814|\n",
      "|          INVST|              20094|         20470|0.9816316560820714|\n",
      "|          ZT�L�|              21542|         21965|0.9807420896881403|\n",
      "+---------------+-------------------+--------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "answer = (\n",
    "    full_log.groupby(\"LogIdentifierID\")\n",
    "    .agg(\n",
    "        F.sum(F.when(F.trim(F.col(\"ProgramClassCD\")).isin([\"COM\",\"PRC\",\"PGI\",\"PRO\",\"LOC\",\"SPO\",\"MER\",\"SOL\"]),\n",
    "                    F.col(\"duration_seconds\")\n",
    "                    ).otherwise(0)\n",
    "             ).alias(\"duration_commercial\"),\n",
    "        F.sum(\"duration_seconds\").alias(\"duration_total\")\n",
    "    )\n",
    "    .withColumn(\"commercial_ratio\",F.col(\"duration_commercial\")/F.col(\"duration_total\"))\n",
    ")\n",
    "answer.orderBy(\"commercial_ratio\",ascending=False).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "46877c78-9d92-4452-a956-38b6073a441b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "324"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "answer.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "c5b9f88d-d914-4799-9d8d-ed443d261502",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "322"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "answer.dropna().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a3580e4b-2759-4bd3-893d-7d7acf7dc58b",
   "metadata": {},
   "source": [
    "exercise 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "13d0b017-9982-44d3-9617-9828a81781db",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-------------+--------------------+\n",
      "|LogIdentifierID|UndertakingNO|    Undertaking_Name|\n",
      "+---------------+-------------+--------------------+\n",
      "|          BRAVO|    315413740|              Bravo!|\n",
      "|           CBET|    320495005|Canadian Broadcas...|\n",
      "|          CFTV3|    535418075|Southshore Broadc...|\n",
      "|           CHEX|    320468002|591987 B.C. Ltd.,...|\n",
      "|           CICT|    430447003|Corus Television ...|\n",
      "|           CIMT|    220450035|Télé Inter-Rives ...|\n",
      "|           CKMI|    210497004|Corus Television ...|\n",
      "|          FIGHT|    535418348|      Fight Network |\n",
      "|         SCSD02|    535422406|Super Channel (fo...|\n",
      "|          SMITH|    535422357|Smithsonian Chann...|\n",
      "+---------------+-------------+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csv_path = os.path.join(folder_path,\"Call_Signs.csv\")\n",
    "df = spark.read.csv(csv_path,\n",
    "            sep = \",\",\n",
    "            header = True,\n",
    "            inferSchema = True,)\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "e212df31-f994-4e26-baa4-4950ed5dcc24",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ProgramClassID: integer (nullable = true)\n",
      " |-- CategoryID: integer (nullable = true)\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- BroadcastLogID: integer (nullable = true)\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- SequenceNO: integer (nullable = true)\n",
      " |-- AudienceTargetAgeID: integer (nullable = true)\n",
      " |-- AudienceTargetEthnicID: integer (nullable = true)\n",
      " |-- ClosedCaptionID: integer (nullable = true)\n",
      " |-- CountryOfOriginID: integer (nullable = true)\n",
      " |-- DubDramaCreditID: integer (nullable = true)\n",
      " |-- EthnicProgramID: integer (nullable = true)\n",
      " |-- ProductionSourceID: integer (nullable = true)\n",
      " |-- FilmClassificationID: integer (nullable = true)\n",
      " |-- ExhibitionID: integer (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- EndTime: string (nullable = true)\n",
      " |-- LogEntryDate: date (nullable = true)\n",
      " |-- ProductionNO: string (nullable = true)\n",
      " |-- ProgramTitle: string (nullable = true)\n",
      " |-- StartTime: string (nullable = true)\n",
      " |-- Subtitle: string (nullable = true)\n",
      " |-- NetworkAffiliationID: integer (nullable = true)\n",
      " |-- SpecialAttentionID: integer (nullable = true)\n",
      " |-- BroadcastOriginPointID: integer (nullable = true)\n",
      " |-- CompositionID: integer (nullable = true)\n",
      " |-- Producer1: string (nullable = true)\n",
      " |-- Producer2: string (nullable = true)\n",
      " |-- Language1: integer (nullable = true)\n",
      " |-- Language2: integer (nullable = true)\n",
      " |-- duration_seconds: integer (nullable = true)\n",
      " |-- LogIdentifierID: string (nullable = true)\n",
      " |-- PrimaryFG: integer (nullable = true)\n",
      " |-- CategoryCD: string (nullable = true)\n",
      " |-- Category_Description: string (nullable = true)\n",
      " |-- ProgramClassCD: string (nullable = true)\n",
      " |-- ProgramClass_Description: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "full_log.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "76076c9b-9e83-4036-acf5-327688f9f1bc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- LogIdentifierID: string (nullable = true)\n",
      " |-- ProgramClassID: integer (nullable = true)\n",
      " |-- CategoryID: integer (nullable = true)\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- BroadcastLogID: integer (nullable = true)\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- SequenceNO: integer (nullable = true)\n",
      " |-- AudienceTargetAgeID: integer (nullable = true)\n",
      " |-- AudienceTargetEthnicID: integer (nullable = true)\n",
      " |-- ClosedCaptionID: integer (nullable = true)\n",
      " |-- CountryOfOriginID: integer (nullable = true)\n",
      " |-- DubDramaCreditID: integer (nullable = true)\n",
      " |-- EthnicProgramID: integer (nullable = true)\n",
      " |-- ProductionSourceID: integer (nullable = true)\n",
      " |-- FilmClassificationID: integer (nullable = true)\n",
      " |-- ExhibitionID: integer (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- EndTime: string (nullable = true)\n",
      " |-- LogEntryDate: date (nullable = true)\n",
      " |-- ProductionNO: string (nullable = true)\n",
      " |-- ProgramTitle: string (nullable = true)\n",
      " |-- StartTime: string (nullable = true)\n",
      " |-- Subtitle: string (nullable = true)\n",
      " |-- NetworkAffiliationID: integer (nullable = true)\n",
      " |-- SpecialAttentionID: integer (nullable = true)\n",
      " |-- BroadcastOriginPointID: integer (nullable = true)\n",
      " |-- CompositionID: integer (nullable = true)\n",
      " |-- Producer1: string (nullable = true)\n",
      " |-- Producer2: string (nullable = true)\n",
      " |-- Language1: integer (nullable = true)\n",
      " |-- Language2: integer (nullable = true)\n",
      " |-- duration_seconds: integer (nullable = true)\n",
      " |-- PrimaryFG: integer (nullable = true)\n",
      " |-- CategoryCD: string (nullable = true)\n",
      " |-- Category_Description: string (nullable = true)\n",
      " |-- ProgramClassCD: string (nullable = true)\n",
      " |-- ProgramClass_Description: string (nullable = true)\n",
      " |-- Undertaking_Name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "full_log.join(df.select(\"LogIdentifierID\",\"Undertaking_Name\"),\n",
    "             on=\"LogIdentifierID\",\n",
    "             how = \"inner\").printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a5bfcef0-26d1-43a3-920c-6be8d5fbd6bb",
   "metadata": {},
   "source": [
    "exercise 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "0b640cdc-f59a-4034-8c9a-fbb5c5e66a6b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------------------+----------------+\n",
      "|ProgramClassCD|ProgramClass_Description|duration_seconds|\n",
      "+--------------+------------------------+----------------+\n",
      "|          PGR |                 PROGRAM|            7200|\n",
      "|          PRC |    PROMOTION OF UPCO...|              30|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              30|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "|          COM |      COMMERCIAL MESSAGE|              15|\n",
      "+--------------+------------------------+----------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    full_log.select(\"ProgramClassCD\",\"ProgramClass_Description\",\"duration_seconds\")\n",
    "    .show(10)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "id": "4f86bbe4-5d8f-4e2d-87ec-e46c95a2afff",
   "metadata": {},
   "outputs": [],
   "source": [
    "full_log_2 = (\n",
    "    full_log#.select(\"ProgramClassCD\",\"ProgramClass_Description\",\"duration_seconds\")\n",
    "    .withColumn(\n",
    "        \"duration_seconds_prc\",F.when(F.trim(F.col(\"ProgramClassCD\"))==\"PRC\",F.expr(\"duration_seconds*0.75\")).otherwise(F.col(\"duration_seconds\"))\n",
    "        )\n",
    ") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "id": "b824356e-88c9-4e57-8ae5-f25bb9880015",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ProgramClassID: integer (nullable = true)\n",
      " |-- CategoryID: integer (nullable = true)\n",
      " |-- LogServiceID: integer (nullable = true)\n",
      " |-- BroadcastLogID: integer (nullable = true)\n",
      " |-- LogDate: date (nullable = true)\n",
      " |-- SequenceNO: integer (nullable = true)\n",
      " |-- AudienceTargetAgeID: integer (nullable = true)\n",
      " |-- AudienceTargetEthnicID: integer (nullable = true)\n",
      " |-- ClosedCaptionID: integer (nullable = true)\n",
      " |-- CountryOfOriginID: integer (nullable = true)\n",
      " |-- DubDramaCreditID: integer (nullable = true)\n",
      " |-- EthnicProgramID: integer (nullable = true)\n",
      " |-- ProductionSourceID: integer (nullable = true)\n",
      " |-- FilmClassificationID: integer (nullable = true)\n",
      " |-- ExhibitionID: integer (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- EndTime: string (nullable = true)\n",
      " |-- LogEntryDate: date (nullable = true)\n",
      " |-- ProductionNO: string (nullable = true)\n",
      " |-- ProgramTitle: string (nullable = true)\n",
      " |-- StartTime: string (nullable = true)\n",
      " |-- Subtitle: string (nullable = true)\n",
      " |-- NetworkAffiliationID: integer (nullable = true)\n",
      " |-- SpecialAttentionID: integer (nullable = true)\n",
      " |-- BroadcastOriginPointID: integer (nullable = true)\n",
      " |-- CompositionID: integer (nullable = true)\n",
      " |-- Producer1: string (nullable = true)\n",
      " |-- Producer2: string (nullable = true)\n",
      " |-- Language1: integer (nullable = true)\n",
      " |-- Language2: integer (nullable = true)\n",
      " |-- duration_seconds: integer (nullable = true)\n",
      " |-- LogIdentifierID: string (nullable = true)\n",
      " |-- PrimaryFG: integer (nullable = true)\n",
      " |-- CategoryCD: string (nullable = true)\n",
      " |-- Category_Description: string (nullable = true)\n",
      " |-- ProgramClassCD: string (nullable = true)\n",
      " |-- ProgramClass_Description: string (nullable = true)\n",
      " |-- duration_seconds_prc: decimal(13,2) (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "full_log_2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "23d0a110-d059-4c9d-800d-424e370ff730",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-------------------+--------------+-------------------+\n",
      "|LogIdentifierID|duration_commercial|duration_total|   commercial_ratio|\n",
      "+---------------+-------------------+--------------+-------------------+\n",
      "|           CIMT|             775.00|           775|1.00000000000000000|\n",
      "|          HPITV|              13.00|            13|1.00000000000000000|\n",
      "|           MSET|            2651.25|          2700|0.98194444444444444|\n",
      "|         TELENO|           17291.25|         17790|0.97196458684654300|\n",
      "|          TLNSP|           14872.50|         15480|0.96075581395348837|\n",
      "|            TV5|           10691.50|         11220|0.95289661319073084|\n",
      "|           CIVM|           11238.25|         11802|0.95223267242840197|\n",
      "|          INVST|           19390.50|         20470|0.94726428920371275|\n",
      "|         ONEBMS|           17425.50|         18522|0.94080012957563978|\n",
      "|       CANALVIE|           19905.00|         21309|0.93411234689567788|\n",
      "+---------------+-------------------+--------------+-------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "answer = (\n",
    "    full_log_2.groupby(\"LogIdentifierID\")\n",
    "    .agg(\n",
    "        F.sum(F.when(F.trim(F.col(\"ProgramClassCD\")).isin([\"COM\",\"PRC\",\"PGI\",\"PRO\",\"LOC\",\"SPO\",\"MER\",\"SOL\"]),\n",
    "                    F.col(\"duration_seconds_prc\")\n",
    "                    ).otherwise(0)\n",
    "             ).alias(\"duration_commercial\"),\n",
    "        F.sum(\"duration_seconds\").alias(\"duration_total\")\n",
    "    )\n",
    "    .withColumn(\"commercial_ratio\",F.col(\"duration_commercial\")/F.col(\"duration_total\"))\n",
    ")\n",
    "answer.orderBy(\"commercial_ratio\",ascending=False).show(10)\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "c761546f-2a7c-478d-901f-e7bf380fd4dd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+------------------+\n",
      "|commercial_ratio|number_of_channels|\n",
      "+----------------+------------------+\n",
      "|             1.0|                 7|\n",
      "|             0.9|                15|\n",
      "|             0.8|                 6|\n",
      "|             0.7|                 1|\n",
      "|             0.5|                 1|\n",
      "|             0.4|                 4|\n",
      "|             0.3|                35|\n",
      "|             0.2|               149|\n",
      "|             0.1|                57|\n",
      "|             0.0|                49|\n",
      "+----------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    answer.fillna(0).groupby(F.round(F.col(\"commercial_ratio\"),1).alias(\"commercial_ratio\"))\n",
    "    .agg(F.count(F.col(\"commercial_ratio\")).alias(\"number_of_channels\"))\n",
    "    .orderBy(\"commercial_ratio\",ascending=False)\n",
    ").show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c354aed3-9d99-49f9-b4f4-dd5f6cc403c1",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
